package org.jivesoftware.openfire.plugin.kafka;

import org.apache.kafka.clients.producer.*;
import org.jivesoftware.util.JiveGlobals;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class ChatLogsProducer {

    private static KafkaProducer<String, String> producer;

    private static final Logger log = LoggerFactory.getLogger(ChatLogsProducer.class);

    public ChatLogsProducer() {
    }

    static {
        log.info("@@@@@@@@ kafka 生产者初始化开始");
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, JiveGlobals.getProperty(KafkaUtil.KAFKA_BOOTSTRAP_SERVERS));
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, JiveGlobals.getIntProperty(KafkaUtil.KAFKA_BUFFER_MEMORY, 1048576));
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, JiveGlobals.getIntProperty(KafkaUtil.KAFKA_BATCH_SIZE, 1024 * 10));
        props.put(ProducerConfig.LINGER_MS_CONFIG, JiveGlobals.getIntProperty(KafkaUtil.KAFKA_LINGER_MS, 1000));

        try {
            producer = new KafkaProducer<>(props);
            log.info("@@@@@@@@ kafka 生产者初始化成功");
        }catch (Exception e){
            log.info("@@@@@@@@ kafka 生产者初始化失败");
            e.printStackTrace();
        }

    }

    public static void send(String title, String logs) {
        try {
            producer.send(new ProducerRecord<String,String>(title, logs),new Callback(){
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (null == e) {
                        log.debug("@@@@@@@@@@ kafka message send success: {}", recordMetadata.toString());
                    } else {
                        log.error("@@@@@@@@@@ kafka message send fail: {}", e);
                    }
                }
            });
        } catch (Exception e) {
            log.error("@@@@@@@@@ producer 发送错误 : ", e);
        }
    }

}
